package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author: Felix
 * Date: 2022/6/13
 * Desc: 日志数据分流
 * 需要启动的进程
 *      zk、kafka、[hdfs]、flume(f1.sh)、DwdTrafficBaseLogSplit
 * 程序执行的流程
 *      模拟前端埋点产生日志数据并进行落盘
 *      flume将磁盘文件上的日志采集到kakfa的topic_log主题中
 *      DwdTrafficBaseLogSplit从kafka的topic_log中读取日志数据
 *          -对脏数据进行处理，将脏数据放到侧输出流发送到kafka的dirty_data主题中
 *          -对日志的新老访客标记进行修复 --- 状态编程
 *          -使用flink侧输出流完成分流操作
 *              错误日志---错误侧输出流
 *              启动日志---启动侧输出流
 *              曝光日志---曝光侧输出流
 *              动作日志---动作侧输出流
 *              页面日志---页面侧输出流
 *          -将不同流的数据写到kafka的dwd主题中，作为事实表
 */
public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) throws Exception {
        //TODO 1. 基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        /*
        //TODO 2. 检查点相关设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/xxxx");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        //TODO 3.从kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "topic_log";
        String groupId = "dwd_traffic_base_log_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //kafkaStrDS.print(">>");

        //TODO 4.对读取的数据进行类型转换以及ETL，将脏数据写到侧输出流   jsonStr->jsonObj
        //4.1 定义侧输出流标签
        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag") {
        };
        //4.2 转换
        SingleOutputStreamOperator<JSONObject> cleanedDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    try {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    } catch (Exception e) {
                        //如果在转换的过程中发生了异常，说明json字符串格式不正确，属于脏数据，放到侧输出流中
                        ctx.output(dirtyTag, jsonStr);
                    }

                }
            }
        );
        //4.3 将侧输出流脏数据写到kafka主题
        //cleanedDS.print(">>");
        DataStream<String> dirtyDS = cleanedDS.getSideOutput(dirtyTag);
        //dirtyDS.print("$$$$$");
        dirtyDS.addSink(MyKafkaUtil.getKafkaProducer("dirty_data"));


        //TODO 5.对新老访客标记进行修复----flink状态编程
        //5.1 按照mid进行分组
        KeyedStream<JSONObject, String> keyedDS = cleanedDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //5.2 修复
        SingleOutputStreamOperator<JSONObject> fixedDS = keyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                //注意：状态的初始化不能在声明的时候进行，因为这个时候还获取不到运行时上下文
                private ValueState<String> lastVisitDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastVisitDateState
                        = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDateState",String.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //获取流中处理的元素新老访客标记
                    String isNew = jsonObj.getJSONObject("common").getString("is_new");
                    //获取状态中存放的上次访问日期
                    String lastVisitDate = lastVisitDateState.value();
                    Long ts = jsonObj.getLong("ts");
                    String curVisitDate = DateFormatUtil.toDate(ts);

                    if("1".equals(isNew)){
                        //前端传递过来的是新访客标记 is_new = 1
                        if(StringUtils.isEmpty(lastVisitDate)){
                            //说明当前设备第一次访问网站  属于新访客  ，标记不需要修复  将这次访问的日期更新到状态中保存起来
                            lastVisitDateState.update(curVisitDate);
                        }else{
                            if(!curVisitDate.equals(lastVisitDate)){
                                //如果状态中的上次访问日期不为空，并且和今天访问日期不相等，说明这个设备曾经访问过，属于老访客，需要将is_new修复为0
                                isNew = "0";
                                jsonObj.getJSONObject("common").put("is_new",isNew);
                            }
                        }
                    }else{
                        //前端传递过来的是老访客标记 is_new = 0
                        if(StringUtils.isEmpty(lastVisitDate)){
                            //说明当前设备的访问是老访客(曾经访问过),但是我们状态中却没有记录曾经访问的日期，我们补充一下
                            String yesterDayStr = DateFormatUtil.toDate(ts - 24 * 3600 * 1000);
                            lastVisitDateState.update(yesterDayStr);
                        }
                    }

                    //将修复之后的jsonObj继续向下游传递
                    out.collect(jsonObj);
                }
            }
        );

        //fixedDS.print(">>>>");

        //TODO 6.将不同类型的数据  放到不同的流中  -----flink侧输出流
        //错误日志-错误侧输出流    启动日志-启动侧输出流  曝光日志-曝光侧输出流  动作日志-动作侧输出流  页面日志-主流
        //6.1 定义侧输出流标签
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        //6.2 分流
        SingleOutputStreamOperator<String> pageDS = fixedDS.process(
            new ProcessFunction<JSONObject, String>() {
                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                    JSONObject errJsonObj = jsonObj.getJSONObject("err");
                    //~~~~~~~错误日志处理~~~~~~~~
                    if (errJsonObj != null) {
                        //如果错误属性不为空，说明当前行为发生了错误，将错误日志保存错误侧输出流
                        ctx.output(errTag, jsonObj.toJSONString());
                        jsonObj.remove("err");
                    }

                    JSONObject startJsonObj = jsonObj.getJSONObject("start");
                    if (startJsonObj != null) {
                        //~~~~~~~启动日志处理,将启动日志放到启动侧输出流~~~~~~~~
                        ctx.output(startTag, jsonObj.toJSONString());
                    } else {
                        //~~~~~~~页面日志处理,曝光日志、动作日志也属于页面日志~~~~~~~~
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        Long ts = jsonObj.getLong("ts");

                        //~~~~~~~曝光日志处理,将曝光日志放到曝光侧输出流~~~~~~~~
                        JSONArray displayArr = jsonObj.getJSONArray("displays");
                        if (displayArr != null) {
                            //遍历当前页面中曝光的所有数据
                            for (int i = 0; i < displayArr.size(); i++) {
                                //获取一条曝光信息
                                JSONObject displayJsonObj = displayArr.getJSONObject(i);
                                //创建一个用于封装完整曝光信息的json对象
                                JSONObject resDisplayJsonObj = new JSONObject();
                                resDisplayJsonObj.put("common", commonJsonObj);
                                resDisplayJsonObj.put("page", pageJsonObj);
                                resDisplayJsonObj.put("ts", ts);
                                resDisplayJsonObj.put("display", displayJsonObj);
                                ctx.output(displayTag, resDisplayJsonObj.toJSONString());
                            }
                        }

                        //~~~~~~~动作日志处理,将动作日志放到动作侧输出流~~~~~~~~
                        JSONArray actionArr = jsonObj.getJSONArray("actions");
                        if (actionArr != null) {
                            for (int i = 0; i < actionArr.size(); i++) {
                                JSONObject actionJsonObj = actionArr.getJSONObject(i);
                                JSONObject resActionJsonObj = new JSONObject();
                                resActionJsonObj.put("common", commonJsonObj);
                                resActionJsonObj.put("page", pageJsonObj);
                                resActionJsonObj.put("action", actionJsonObj);
                                ctx.output(actionTag, resActionJsonObj.toJSONString());
                            }
                        }

                        //页面日志  放到主流
                        jsonObj.remove("displays");
                        jsonObj.remove("actions");
                        out.collect(jsonObj.toJSONString());
                    }
                }
            }
        );

        DataStream<String> errDS = pageDS.getSideOutput(errTag);
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);

        pageDS.print(">>>>");
        errDS.print("~~~");
        startDS.print("@@@");
        displayDS.print("###");
        actionDS.print("$$$");

        //TODO 7.将不同流的数据写到kafka的dwd不同主题中
        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log"));
        errDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_err_log"));
        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        actionDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));

        env.execute();
    }
}
